import rx.Action;

public class AsyncJob<T> {

    Action<Callback<T>> action;

    NewThreadWorker worker = new NewThreadWorker("subscribeOn");
    NewThreadWorker worker2 = new NewThreadWorker("ObserveOn");

    public static <T> AsyncJob<T> create(Action<Callback<T>> action) {
        return new AsyncJob(action);
    }

    public static <T> AsyncJob<T> create(Action0<T> action0) {
        return AsyncJob.create(new Action<Callback<T>>() {
            @Override
            public void call(Callback<T> callback) {
                T r = action0.call();
                callback.onNext(r);
                callback.onComplete();
            }
        });
    }

    public static <T> AsyncJob<T> just(Iterable<T> iterable) {
        return AsyncJob.create(new Action<Callback<T>>() {
            @Override
            public void call(Callback<T> callback) {
                for (T anIterable : iterable) {
                    callback.onNext(anIterable);
                }
                callback.onComplete();
            }
        });
    }

    private AsyncJob(Action<Callback<T>> action) {
        this.action = action;
    }

    public void start(Callback<T> callback) {
        action.call(callback);
    }

    public AsyncJob<T> createOnNewThread() {
        AsyncJob<T> upstream = this;
        return new AsyncJob<T>(new Action<Callback<T>>() {
            @Override
            public void call(Callback<T> callback) {
                worker2.schedule(new Runnable() {
                    @Override
                    public void run() {
                        upstream.start(new Callback<T>() {
                            @Override
                            public void onNext(T t) {
                                callback.onNext(t);
                            }

                            @Override
                            public void onComplete() {
                                callback.onComplete();
                            }

                            @Override
                            public void onError(Throwable e) {
                                callback.onError(e);
                            }
                        });
                    }
                });
                worker2.finish();
            }
        });
    }

    public AsyncJob<T> callbackOnNewThread() {
        AsyncJob<T> upstream = this;
        return new AsyncJob<>(new Action<Callback<T>>() {
            @Override
            public void call(Callback<T> callback) {
                upstream.start(new Callback<T>() {
                    @Override
                    public void onNext(T t) {
                        worker.schedule(new Runnable() {
                            @Override
                            public void run() {
                                callback.onNext(t);
                            }
                        });
                    }
                    @Override
                    public void onComplete() {
                        worker.schedule(new Runnable() {
                            @Override
                            public void run() {
                                callback.onComplete();
                            }
                        });
                        worker.finish();
                    }

                    @Override
                    public void onError(Throwable e) {
                        worker.schedule(new Runnable() {
                            @Override
                            public void run() {
                                callback.onError(e);
                            }
                        });
                        worker.finish();
                    }
                });
            }
        });

    }

    public <R> AsyncJob<R> map(Func<T, R> func) {
        final AsyncJob<T> upstream = this;
        return new AsyncJob<R>(new Action<Callback<R>>() {
            @Override
            public void call(Callback<R> callback) {
                upstream.start(new Callback<T>() {
                    @Override
                    public void onNext(T t) {
                        callback.onNext(func.call(t));
                    }

                    @Override
                    public void onComplete() {
                        callback.onComplete();
                    }

                    @Override
                    public void onError(Throwable e) {
                        callback.onError(e);
                    }
                });
            }
        });
    }


//
//    public <R> AsyncJob<R> flatmap(Func<T, AsyncJob<R>> func) {
//        final AsyncJob<T> sourse = this;
//        return new AsyncJob<R>() {
//            @Override
//            public void start(Callback<R> callback) {
//                sourse.start(new Callback<T>() {
//                    @Override
//                    public void onResult(T t) {
//                        AsyncJob<R> map = func.call(t);
//                        map.start(new Callback<R>() {
//                            @Override
//                            public void onResult(R r) {
//                                callback.onResult(r);
//                            }
//
//                            @Override
//                            public void onError(Exception e) {
//                                callback.onError(e);
//                            }
//                        });
//                    }
//
//                    @Override
//                    public void onError(Exception e) {
//                        callback.onError(e);
//                    }
//                });
//            }
//        };
//    }

}
